defmodule SockWorkstation.GrpcVADClient do
  use GenServer
  require Logger
  alias Vad.VADService.Stub
  alias Vad.{SpeechTimestamps, Timestamp, AudioChunk}

  # gRPC 服务器地址
  @grpc_server "localhost:50052"

  @name __MODULE__

  alias SockWorkstation.{GrpcVADClient, ASRHelperGen}

  defstruct channel: nil,
            stream_ref: nil,
            asr_helper_pid: nil,
            timestamps: [],
            session_id: nil,
            is_streaming?: false,
            self_pid: nil,
            bin_buffer: <<>>

  # 客户端 API

  def start_link(session_id) do
    GenServer.start_link(@name, session_id, name: via_tuple(session_id, "vad_helper"))
  end

  def get_state(session_id) do
    GenServer.call(via_tuple(session_id, "vad_helper"), :get_state)
  end

  def send_pcm(session_id, audio_data) do
    GenServer.cast(via_tuple(session_id, "vad_helper"), {:sent_pcm, audio_data})
  end

  # GenServer 回调

  @impl true
  def init(session_id) do
    with {:ok, channel} <- GRPC.Stub.connect(@grpc_server),
         {:ok, asr_helper_pid} <- ASRHelperGen.fetch_asr_helper_pid(session_id) do
      # 初始化 gRPC 连接
      self_pid = self()
      # {:ok, channel} = GRPC.Stub.connect(@grpc_server)
      stream = Stub.stream_vad(channel)
      # schedule_chunk_vad(self_pid)
      IO.puts("Grpc VAD Client Connected!")

      {:ok,
       %GrpcVADClient{
         channel: channel,
         stream_ref: stream,
         asr_helper_pid: asr_helper_pid,
         timestamps: [],
         session_id: session_id,
         is_streaming?: true,
         self_pid: self_pid,
         bin_buffer: <<>>
       }}
    else
      _ ->
        {:ok, %GrpcVADClient{self_pid: self()}}
    end
  end

  def connect(session_id) do
    GenServer.cast(via_tuple(session_id, "vad_helper"), :connect)
  end

  # def add_asr_helper_pid(session_id, pid) do
  #   GenServer.cast(via_tuple(session_id, "vad_helper"), {:add_asr_helper_pid, pid})
  # end

  @impl true
  def handle_call(:get_state, _from, state) do
    {:reply, state, state}
  end

  @impl true
  def handle_cast(:connect, state) do
    with {:ok, channel} <- GRPC.Stub.connect(@grpc_server),
         {:ok, asr_helper_pid} <- ASRHelperGen.fetch_asr_helper_pid(state.session_id) do
      stream = Stub.stream_vad(channel)

      {:noreply,
       %GrpcVADClient{state | channel: channel, stream_ref: stream, is_streaming?: true, asr_helper_pid: asr_helper_pid}}
    end
  end

  # def handle_cast(:connect, state) do
  #   {:noreply, state}
  # end

  # def handle_cast({:add_asr_pid, asr_helper_pid}, state) when asr_helper_pid == state.asr_helper_pid do
  #   {:noreply, state}
  # end

  # def handle_cast({:add_asr_helper_pid, asr_helper_pid}, state) do
  #   {:noreply, %{state | asr_helper_pid: asr_helper_pid}}
  # end

  def handle_cast({:sent_pcm, pcm_data}, state) do
    new_state = process_audio_chunk(state, pcm_data)
    {:noreply, new_state}
  end

  @impl true

  def handle_info({:gun_response, _conn_pid, _stream_ref, :fin, status, _headers}, state) do
    # 处理 gRPC 流结束的情况
    IO.puts("Received gRPC stream response with status: #{status}")
    {:noreply, state}
  end

  def handle_info({:gun_response, _conn_pid, _stream_ref, :nofin, status, _headers}, state) do
    # 处理 gRPC 流数据的情况
    IO.puts("Received gRPC stream data with status: #{status}")
    {:noreply, state}
  end

  def handle_info({:gun_trailers, _pid, _ref, trailers}, state) do
    # 处理 trailers，例如打印错误信息
    trailers
    |> Enum.each(fn {key, value} ->
      Logger.error("GRPC Trailer: #{key} = #{value}")
    end)

    {:noreply, state}
  end

  def handle_info({:gun_data, _conn_pid, _stream_ref, :fin, data}, state) do
    # 处理流结束时的数据
    IO.puts("Received final data chunk: #{inspect(data)}")
    {:noreply, state}
  end

  def handle_info({:gun_data, _conn_pid, _stream_ref, :nofin, data}, state) do
    # 处理流中的数据

    case SpeechTimestamps.handle_grpc_message(data) do
      {:ok, parsed_data} ->
        [%Timestamp{start: start_stamps, end: end_stamps}] = parsed_data.timestamps
        # state = Map.put(state, :timestamps, parsed_data.timestamps)
       start_stps=:erlang.float_to_binary(start_stamps, decimals: 0)
       end_stps=:erlang.float_to_binary(end_stamps, decimals: 0)
        stamps = [
          start_stps,
           end_stps
        ]
        IO.puts("Received data chunk: #{inspect(stamps)}")
         if String.to_integer(end_stps) > 0, do: send(state.asr_helper_pid, :vad_done)
        {:noreply, Map.put(state, :timestamps, stamps)}

      {:error, reason} ->
        IO.puts("Failed to decode data: #{inspect(reason)}")
        {:noreply, state}
    end
  end

  def handle_info({:gun_error, _conn_pid, _stream_ref, reason}, state) do
    # 处理错误
    IO.puts("Error in gRPC stream: #{inspect(reason)}")
    {:noreply, state}
  end

  def handle_info({:gun_down, _conn_pid, _stream_ref, reason}, state) do
    # 处理错误
    IO.puts("down in gRPC stream: #{inspect(reason)}")
    {:noreply, state}
  end

  def handle_info(msg, state) do
    Logger.warning("Unhandled message:", [inspect(msg)])
    {:noreply, state}
  end

  defp process_audio_chunk(state, pcm_data) when state.is_streaming? == true do
    session_id = state.session_id

    padded_chunks = padding_pcm_data(pcm_data)

    Enum.each(padded_chunks, fn chunk ->
      audio_chunk = %AudioChunk{
        audio_data: chunk,
        session_id: session_id
      }

      GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: false)
    end)

    state
  end

  defp process_audio_chunk(state, pcm_data) do
    session_id = state.session_id
    stream_ref = Stub.stream_vad(state.channel)

    padded_chunks = padding_pcm_data(pcm_data)

    Enum.each(padded_chunks, fn chunk ->
      audio_chunk = %AudioChunk{
        audio_data: chunk,
        session_id: session_id
      }

      GRPC.Stub.send_request(stream_ref, audio_chunk, end_stream: false)
    end)

    %{state | stream_ref: stream_ref, is_streaming?: true}
  end

  defp padding_pcm_data(data) when byte_size(data) == 1920 do
    # 分割数据为两段，每段 960 字节
    <<chunk1::binary-size(960), chunk2::binary-size(960)>> = data

    # 创建 32 字节的静音数据（零字节）
    padding = :binary.copy(<<0>>, 32)

    # 在每段前后添加静音数据
    padded_chunk1 = <<padding::binary, chunk1::binary, padding::binary>>
    padded_chunk2 = <<padding::binary, chunk2::binary, padding::binary>>

    [padded_chunk1, padded_chunk2]
  end

  defp padding_pcm_data(_), do: raise("Input data must be 1920 bytes")

  defp via_tuple(session_id, type) do
    {:via, Registry, {SockWorkstation.SessionRegistry, {session_id, type}}}
  end
end

# try do
#   result1 = SpeechTimestamps.decode(data)
#   # result2 = SpeechTimestamps.decode(binary_data2)

#   IO.inspect(result1, label: "Decoded SpeechTimestamps 1")
#   # IO.inspect(result2, label: "Decoded SpeechTimestamps 2")
# rescue
#   e in Protobuf.DecodeError ->
#     IO.puts("Failed to decode data: #{inspect(e)}")
# end

# def pad_data(data, chunk_size \\ 1024) do
#   # 确保数据长度为 960 字节
#   if byte_size(data) != 960 do
#     raise "Invalid data size. Expected 960 bytes, got #{byte_size(data)} bytes."
#   end

#   # 前后各添加 32 字节的填充数据（使用零字节填充）
#   padded_data = <<0::size(32 * 8), data::binary, 0::size(32 * 8)>>

#   # 确保生成的数据长度为 1024 字节
#   if byte_size(padded_data) != chunk_size do
#     raise "Padded data size mismatch. Expected #{chunk_size} bytes, got #{byte_size(padded_data)} bytes."
#   end

#   padded_data
# end

# defp schedule_chunk_vad(pid) do
#   Process.send_after(pid, :chunk_vad, 60)
# end

# def split_data(data, size) when byte_size(data) > size do
#   <<chunk::binary-size(size), remaining::binary>> = data
#   {chunk, remaining}
# end

# def split_data(data, _size) do
#   # padded_data = data <> :binary.copy(<<0>>, 1024 - byte_size(data))
#   # string_data = :binary.bin_to_list(data)
#   # IO.puts(byte_size(padded_data))
#   {data, <<>>}
# end
# defp process_audio_chunk(state, pcm_data) do
#   # bin_buffer = <<state.bin_buffer::binary, pcm_data::binary>>
#   padded_chunks = padding_pcm_data(pcm_data)
#   # 按需分割音频数据并发送
#   # {chunk, remaining} = split_data(bin_buffer, 1024)
#   # if remaining != <<>> do
#   Enum.each(padded_chunks, fn chunk ->
#     audio_chunk = %AudioChunk{
#       audio_data: chunk,
#       session_id: state.session_id
#     }

#     GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: false)
#     # IO.puts("Chunk size: #{byte_size(chunk)} bytes")
#   end)

#   # 发送音频数据到 gRPC 流

#   # Logger.debug("Sent audio chunk to VAD server")
#   # end
#   state
#   # %{state | bin_buffer: remaining}
# end

# def handle_info(:chunk_vad, state) do
#   case split_data(state.bin_buffer, 1024) do
#     {pcm_chunk, <<>>} ->
#       # 如果没有剩余数据，继续等待
#       schedule_chunk_vad(state.self_pid)
#       {:noreply, %{state | bin_buffer: <<>>}}

#     {pcm_chunk, remaining_buffers} ->
#       # 创建音频数据块
#       audio_chunk = %AudioChunk{
#         audio_data: pcm_chunk,
#         session_id: state.session_id
#       }

#       # 如果是第一次发送，启动流式调用
#       # if is_nil(state.stream_ref) do
#         # stream_ref = Stub.stream_vad(state.channel)
#         GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: false)
#         # Logger.info("Started streaming for session: #{state.session_id}")
#         # new_state = %{state | bin_buffer: remaining_buffers, stream_ref: stream_ref, is_streaming?: true}
#       # else
#         # 继续发送音频数据
#         # GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: false)
#         # new_state = %{state | bin_buffer: remaining_buffers}
#       # end

#       schedule_chunk_vad(state.self_pid)
#       {:noreply, %{state | bin_buffer: remaining_buffers, is_streaming?: true}}
#   end
# end
# def handle_info(:chunk_vad, state) do
#   case split_data(state.bin_buffer, 1024) do
#     {pcm_chunk, <<>>} ->
#       schedule_chunk_vad(state.self_pid)
#       {:noreply, %{state | bin_buffer: <<>>}}

#     {pcm_chunk, remaining_buffers} ->
#       schedule_chunk_vad(state.self_pid)

#       audio_chunk = %AudioChunk{
#         audio_data: pcm_chunk,
#         session_id: state.session_id
#       }

#       # 如果是第一次发送，启动流式调用
#       if is_nil(state.stream_ref) do
#         stream_ref = Stub.stream_vad(state.channel)
#       GRPC.Stub.send_request(stream_ref, audio_chunk, end_stream: true)

#       Logger.info("Started streaming for session: #{state.session_id}")
#       {:noreply, %{state | bin_buffer: remaining_buffers, stream_ref: stream_ref, is_streaming?: true}}
#       else
#       # 继续发送音频数据
#       GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: true)
#       {:noreply, state}
#       end
#   end
# end

# def handle_cast({:sent_pcm, pcm_data}, state) do
#    IO.puts("recv_pcm #{byte_size(state.bin_buffer)}")
#   bin_buffer = <<state.bin_buffer::binary, pcm_data::binary>>
#   {:noreply, %{state | bin_buffer: bin_buffer}}
# end

# def handle_info(:run_sent_vad, state) do
#   # 处理服务端返回的语音时间戳
#   {:ok, reply_enum} = GRPC.Stub.recv(state.stream_ref)
#   replies = Enum.map(reply_enum, fn [] -> [] end)

#   Logger.info("Received speech timestamps: #{inspect(replies)}")
#   {:noreply, state}
# end

# def handle_info({:grpc_stream, %SpeechTimestamps{timestamps: timestamps}}, state) do
#   # 处理服务端返回的语音时间戳
#   Logger.info("Received speech timestamps: #{inspect(timestamps)}")
#   {:noreply, state}
# end

# def handle_info(:grpc_stream_start, state) do
#   # 创建 AudioChunk 消息
#   # IO.puts("recv_opus #{byte_size(audio_data)}")
#   audio_chunk = %AudioChunk{
#     audio_data: state.bin_buffer,
#     session_id: state.session_id
#   }

#   # 如果是第一次发送，启动流式调用
#   if is_nil(state.stream_ref) do
#     stream = Stub.stream_vad(state.channel)
#     GRPC.Stub.send_request(stream, audio_chunk, end_stream: true)
#     send(state.self_pid, :run_sent_vad)
#     Logger.info("Started streaming for session: #{state.session_id}")
#     {:noreply, %{state | stream_ref: stream, is_streaming?: true}}
#   else
#     # 继续发送音频数据
#     GRPC.Stub.send_request(state.stream_ref, audio_chunk, end_stream: true)
#     {:noreply, state}
#   end
# end

# def handle_info(:grpc_stream_end, state) do
#   # 流式调用结束
#   Logger.info("Stream ended for session")
#   {:noreply, %{state | reply_enum: []}}
# end

# def handle_info(:grpc_stream_start, state) do
#   Logger.info("Stream started for session")
#   {:noreply, state}
# end

# def handle_info({:grpc_error, error}, state) do
#   # 处理 gRPC 错误
#   Logger.error("gRPC error: #{inspect(error)}")
#   {:noreply, state}
# end
